package com.kalvan.core.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/***
 * RocketMQ 异步消息生产者
 *
 * @author chenliang
 */
@Slf4j
@Component
public class MqProducer {
    private final static String LOG_PREFIX = "生产者";
    private final DefaultMQProducer producer = new DefaultMQProducer(MqProducer.class.getSimpleName());

    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;

    /**
     * 初始化
     */
    @PostConstruct
    public void start() {
        try {
            log.info("MQ：启动{} {}", LOG_PREFIX, namesrvAddr);
            //指定NameServer地址，多个地址以 ; 隔开
            producer.setNamesrvAddr(namesrvAddr);
            //消息发送失败重试次数
            producer.setRetryTimesWhenSendFailed(2);
            //异步发送失败重试次数
            producer.setRetryTimesWhenSendAsyncFailed(0);
            //消息没有发送成功，是否发送到另外一个Broker中
            producer.setRetryAnotherBrokerWhenNotStoreOK(false);
            //自动创建topic
            producer.setCreateTopicKey(MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC);
            //启动生成者调用一次
            producer.start();
        } catch (MQClientException e) {
            log.error("MQ：启动{}失败：{}-{}", LOG_PREFIX, e.getResponseCode(), e.getErrorMessage());
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    /**
     * 发送消息(异步)
     * 异步传输通常用于响应时间敏感的业务场景
     *
     * @param data  消息内容
     * @param topic 主题
     * @param tags  标签
     * @param keys  唯一主键
     */
    public void sendAsyncMessage(String topic, String tags, String keys, String data) {
        try {
            byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);
            Message mqMsg = new Message(topic, tags, keys, messageBody);
            /// 延时的级别为3 对应的时间为10s 就是发送后延时10S在把消息投递出去
            /// mqMsg.setDelayTimeLevel(3);
            producer.send(mqMsg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.debug("MQ: {}发送Async消息成功 {}", LOG_PREFIX, sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("MQ: {}发送Async消息异常keys{}", LOG_PREFIX, keys, throwable);
                }
            });
        } catch (Exception e) {
            log.error("MQ: {}发送Async消息异常", LOG_PREFIX, e);
        }
    }

    /**
     * 发送消息(同步)
     * 可靠的同步传输用于广泛的场景，如重要的通知消息，短信通知，短信营销系统等
     *
     * @param data  消息内容
     * @param topic 主题
     * @param tags  标签
     * @param keys  唯一主键
     * @return SendResult
     */
    public SendResult sendSyncMessage(String topic, String tags, String keys, String data) {
        try {
            byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);
            Message mqMsg = new Message(topic, tags, keys, messageBody);
            /// 延时的级别为3 对应的时间为10s 就是发送后延时10S在把消息投递出去
            /// mqMsg.setDelayTimeLevel(3);
            return producer.send(mqMsg);
        } catch (Exception e) {
            log.error("MQ: {}发送Syn消息异常", LOG_PREFIX, e);
        }
        return null;
    }

    /**
     * 发送消息(单向模式)
     * 单向传输用于需要中等可靠性的情况，例如日志收集
     *
     * @param data  消息内容
     * @param topic 主题
     * @param tags  标签
     * @param keys  唯一主键
     */
    public void sendOnewayMessage(String topic, String tags, String keys, String data) {
        try {
            byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET);
            Message mqMsg = new Message(topic, tags, keys, messageBody);
            /// 延时的级别为3 对应的时间为10s 就是发送后延时10S在把消息投递出去
            /// mqMsg.setDelayTimeLevel(3);
            producer.sendOneway(mqMsg);
        } catch (Exception e) {
            log.error("MQ: {}发送Oneway消息异常", LOG_PREFIX, e);
        }
    }


    @PreDestroy
    public void stop() {
        if (producer != null) {
            producer.shutdown();
            log.info("MQ：关闭{} {}", LOG_PREFIX, namesrvAddr);
        }
    }
}
